package com.jacky.iot;

import com.jacky.iot.config.MqttServerConfig;
import com.jacky.iot.handler.MqttHandler;
import com.jacky.iot.handler.MqttHandlerFactory;
import com.jacky.iot.manager.EndpointManager;
import io.vertx.core.Vertx;
import io.vertx.core.json.JsonObject;
import io.vertx.core.net.PemKeyCertOptions;
import io.vertx.core.net.PfxOptions;
import io.vertx.mqtt.MqttServer;
import io.vertx.mqtt.MqttServerOptions;

import java.util.HashSet;

public class VertMqttServer {

    public static void main(String[] args) {
        Vertx vertx = Vertx.vertx();
        JsonObject config = vertx.fileSystem().readFileBlocking("config.json").toJsonObject();
        MqttServerConfig serverConfig = new MqttServerConfig(config);
        EndpointManager endpointManager = new EndpointManager();

        MqttServerOptions options = new MqttServerOptions()
                .setPort(serverConfig.getPort())
                .setHost(serverConfig.getHost());

        if (serverConfig.isEnableSsl()) {
            options.setSsl(true)
                    // pfx方式
//                    .setPfxKeyCertOptions(new PfxOptions()
//                            .setPath(serverConfig.getKeystorePath())
//                            .setPassword(serverConfig.getKeystorePassword()))
                    // pem方式
                    .setKeyCertOptions(new PemKeyCertOptions()
                            .setCertPath("server-cert.pem")
                            .setKeyPath("server-key.pem"))
            ;
        }

        // 根据配置选择协议版本处理器
        MqttHandler handler = MqttHandlerFactory.createHandler(
                serverConfig.getProtocolVersion(),
                endpointManager,
                serverConfig
        );

        MqttServer server = MqttServer.create(vertx, options);
        server.endpointHandler(endpoint -> {
            handler.handleConnection(endpoint);

            endpoint
                    .publishHandler(message -> handler.handlePublish(endpoint, message))
                    .subscribeHandler(subscribe -> handler.handleSubscribe(endpoint, new HashSet<>(subscribe.topicSubscriptions()), subscribe))
                    .unsubscribeHandler(unsubscribe -> handler.handleUnsubscribe(endpoint, new HashSet<>(unsubscribe.topics()), unsubscribe));

            endpoint.accept(false);
        });

        server.listen(result -> {
            if (result.succeeded()) {
                System.out.printf("MQTT Server (%s) started on port %d%n",
                        serverConfig.getProtocolVersion(), serverConfig.getPort());
            } else {
                System.err.println("Failed to start MQTT Server: " + result.cause());
            }
        });
    }
}
